package com.study.bigdata.spark.streaming

import com.study.bigdata.spark.util.JDBCUtil
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer

object SparkStreaming11_Req1_BlackList1 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc =new StreamingContext(sparkConf,Seconds(3))
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "atguigu",//消费者组ID
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    val kafkaDataDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent, //位置策略，采集结点和计算结点如何做匹配？自己匹配：框架匹配
      ConsumerStrategies.Subscribe[String, String](Set("atguiguNew"), kafkaPara) //消费者策略,主题：atguiguNew
    )
    val adClickData = kafkaDataDS.map {
      kafkaData => {
        val data = kafkaData.value()
        val datas = data.split(" ")
        AdClickData(datas(0), datas(1), datas(2), datas(3), datas(4))
      }
    }

    val ds = adClickData.transform(
      rdd => {
        //TODO 通过JDBC周期性获取黑名单数据
        val blockList = ListBuffer[String]()

        val conn = JDBCUtil.getConnection
        val pstat = conn.prepareStatement("select userid from black_list")
        val rs = pstat.executeQuery()
        while (rs.next()) {
          blockList.append(rs.getString(1))
        }
        rs.close()
        pstat.close()
        conn.close()

        //TODO 判断点击用户是否在黑名单中
        val filterRDD = rdd.filter(
          data => {
            !blockList.contains(data.user)
          }
        )
        //TODO 如果点击用户不在黑名单中，那么进行统计数量（每个采集周期）
        filterRDD.map(
          data => {
            val sdf = new SimpleDateFormat("yyyy-MM-dd")
            val day = sdf.format(new Date((data.ts.toLong)))
            val user = data.user
            val ad = data.ad
            ((day, user, ad), 1)
          }
        ).reduceByKey(_ + _)
      }
    )

    ds.foreachRDD(
      rdd =>{
        //rdd.foreach方法每一条数据创建连接，大数据下会浪费资源
        //foreach方法是RDD的算子，算子之外的代码都是在Driver端执行，算子内的代码在Executor端执行
        //这样就会涉及闭包操作，Driver需要将数据传到Executor端就需要序列化
        //而数据库的连接对象是不能序列化的

        //RDD提供了一个算子可以有效提升效率 以一个分区为单位来创建连接对象,可以大幅减少连接对象的数量，提升效率
//        rdd.foreachPartition(
//          iter =>{
//            val conn = JDBCUtil.getConnection
//            iter.foreach{
//              case ((day, user, ad), count) =>{
//
//              }
//            }
//            conn.close()
//          }
//        )
        rdd.foreach{
          case ((day, user, ad), count) =>{
            println(s"${day} ${user} ${ad} ${count}")
            if (count>=30){
              //TODO 如果统计的点击数量超过阈值(30)，那么将用户拉入到黑名单
              val conn = JDBCUtil.getConnection
              val sql = """
                          |insert into black_list (userid) values (?)
                          |on DUPLICATE KEY
                          |UPDATE userid=?
                          |""".stripMargin
              JDBCUtil.executeUpdate(conn,sql,Array(user,user))
              conn.close()

            }else{
              //TODO 如果没有超过阈值，那么需要将当天的广告点击数量进行更新
              val conn = JDBCUtil.getConnection
              val sql = """
                          |select * from user_ad_count
                          |where dt=? and userid=? and adid=?
                          |""".stripMargin
              val fig =JDBCUtil.isExist(conn,sql,Array(day,user,ad))

              //查询统计表数据
              if (fig){
                //如果存在数据，那就更新
                val sql1 = """
                            |update user_ad_count
                            |set count=count + ?
                            |where dt =? and userid =? and adid =?
                            |""".stripMargin
                JDBCUtil.executeUpdate(conn,sql1,Array(conn,day,user,ad))
                //TODO 判断更新后的点击数据是否超过阈值，如果超过，拉入黑名单
                val sql2 = """
                             |select * from user_ad_count
                             |where dt =? and userid =? and adid =? and count >=30
                             |""".stripMargin
                val fig1 =JDBCUtil.isExist(conn,sql2,Array(day,user,ad))

                if(fig1){
                  val sql3 = """
                              |insert into black_list (userid) values (?)
                              |on DUPLICATE KEY
                              |UPDATE userid=?
                              |""".stripMargin
                  JDBCUtil.executeUpdate(conn,sql3,Array(user,user))
                }
              }else{
                val sql4 = """
                             |insert into user_ad_count (dt,userid,adid,count) values (?,?,?,?)
                             |""".stripMargin
                JDBCUtil.executeUpdate(conn,sql4,Array(day,user,ad,count))
              }
              conn.close()
            }
          }
        }
      }
    )

    ssc.start()
    ssc.awaitTermination()
  }
  //广告点击数据样例类
  case class AdClickData(ts: String, area: String, city: String, user: String, ad: String)
}
